'''
Created on Apr 5, 2016

@author: JimmyMo
'''
from writter.writter import Writter
# from pykafka import KafkaClient
# from kafka import KafkaClient, SimpleProducer, SimpleConsumer
import time
from kafka.producer.kafka import KafkaProducer

class KafkaWritter(Writter):
    def _parse_options(self, raw_options):
        self.topic = "logpumper"
        self.host = "localhost"
        self.port = 9092
        self.conn_string = "%s:%s" % (self.host, self.port)
    def __init__(self, queue, raw_options):
        super().__init__(queue, raw_options)
#         self.producer = KafkaProducer(bootstrap_servers=self.conn_string)
#         self.producer = SimpleProducer(kafka, async=True)
        
    def run(self):
        self.producer = KafkaProducer(bootstrap_servers=[self.conn_string], batch_size=10240, acks=0, linger_ms=0)
#         client = KafkaClient(self.conn_string)
#         topic = client.topics[b'helloworld4']
#         self.producer = topic.get_producer()
        
        counter = 0
        now = time.clock()
#         for i in range(200):
#             self.producer.produce((b"habc"))
#             self.producer.send_messages(self.topic, ("habc %d" % i).encode())
        while 1:
            event = self.data.get()
            if (event == "_produced_end"):
                self.producer.flush()
                break;
            counter = counter + 1
            if (counter % 100 == 0):
                print(counter)
                print(time.clock() - now)
                print("there are %d records in queue..." % self.data._qsize())
#                 self.producer.flush()
#             self.producer.send_messages(self.topic, event)
            self.producer.send(self.topic, event)
#             print(counter)
#             print(event)
        pass